package cn.monkey.io.transport.mq.redisson;

import cn.monkey.io.transport.mq.ConcurrentMapExchangeContext;
import cn.monkey.io.transport.core.Exchange;
import cn.monkey.io.transport.core.ExchangeContext;
import io.netty.util.AttributeKey;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

public class SimpleRedissonExchange implements RedissonExchange {
    private final String id;
    private final RedissonInbound inbound;
    private final RedissonOutbound outbound;
    private final AtomicBoolean closed;
    private final ExchangeContext context;

    public SimpleRedissonExchange(String id, RedissonInbound inbound, RedissonOutbound outbound) {
        this.id = id;
        this.inbound = inbound;
        this.outbound = outbound;
        this.closed = new AtomicBoolean(false);
        this.context = new ConcurrentMapExchangeContext();
    }

    @Override
    public String id() {
        return this.id;
    }

    public ExchangeContext context() {
        return this.context;
    }

    @Override
    public Mono<Void> close() {
        if (!this.closed.compareAndSet(false, true)) {
            return Mono.error(new IllegalAccessError("exchange has already closed"));
        }
        List<Mono<Void>> list = List.of(this.inbound().close(), this.outbound().close());
        return Flux.merge(list).then();
    }

    @Override
    public boolean isActive() {
        return !this.closed.get();
    }

    @Override
    public RedissonInbound inbound() {
        return this.inbound;
    }

    @Override
    public RedissonOutbound outbound() {
        return this.outbound;
    }

    @Override
    public RedissonExchange bind() {
        this.setAttribute(AttributeKey.valueOf(Exchange.class, Exchange.KEY).name(), this);
        return this;
    }
}
